package server

/**
* @author aFun
* @data 2025/4/10 01:02
* @email: afunaaa222@163.com / 1030935903
* @address: Guangdong University of Technology, China
 */

import (
	"camp_lucky_draw/entity"
	"camp_lucky_draw/transfer"
	"encoding/json"
	"fmt"
	"github.com/garyburd/redigo/redis"
	"github.com/jinzhu/gorm"
	_ "github.com/jinzhu/gorm/dialects/mysql" // 引入mysql驱动
	"github.com/spf13/viper"
	"github.com/streadway/amqp"
	"log"
	"net"
	"os"
	"sync"
	"time"
)

type UserMgr struct {
	Mutex        sync.RWMutex
	OnlineUsers  map[net.Conn]*entity.Role
	OnlineIdConn map[uint]net.Conn
}

type campaign struct {
	StartTime int64
	EndTime   int64
}

var (
	OnlineUserMgr     *UserMgr
	CampaignData      map[int64]campaign
	CampLuckyDrawData map[string]interface{}
	DB                *gorm.DB
	RedisPool         *redis.Pool
	MQ                *amqp.Connection
)

func InnitOnlineUserMgr() {
	OnlineUserMgr = &UserMgr{
		OnlineUsers:  make(map[net.Conn]*entity.Role),
		OnlineIdConn: make(map[uint]net.Conn),
	}
}

func Init_config() {
	viper.SetConfigName("app")
	viper.AddConfigPath("../config")
	err := viper.ReadInConfig()
	if err != nil {
		panic(err)
	}
	//fmt.Println("config app:", viper.Get("mysql.dns"))

}

func Init_campaign_data() {
	campaignData, err := os.ReadFile("../json_data/campaign_data.json")
	if err != nil {
		panic(err)
	}
	err = json.Unmarshal(campaignData, &CampaignData)
	if err != nil {
		panic(err)
	}
}

func Init_camp_lucky_draw_data() {
	luckyDrawData, err := os.ReadFile("../json_data/camp_lucky_draw_data.json")
	if err != nil {
		panic(err)
	}
	err = json.Unmarshal(luckyDrawData, &CampLuckyDrawData)
	if err != nil {
		panic(err)
	}
}

func InitMYSQL() {
	var err error
	DB, err = gorm.Open("mysql", viper.GetString("mysql.dns"))
	if err != nil {
		panic("failed to connect database,err:" + err.Error())
	}
	DB.LogMode(true)

	// 压测需要：清空所有表的数据
	DB.Exec("TRUNCATE TABLE roles;")
	DB.Exec("TRUNCATE TABLE lucky_draws;")

	DB.AutoMigrate(&entity.Role{}, &entity.LuckyDraw{})

	// 插入新数据
	err = InsertRole()
	if err != nil {
		fmt.Println("insert role err:", err)
	}
}

func InsertRole() error {
	// TODO 优化项：一条一条插入太慢了
	for i := 0; i < 500; i++ {
		tx := DB.Begin()
		role := &entity.Role{
			Lev:    55,
			Asset:  99999,
			Tokens: 0,
		}
		if err := tx.Create(role).Error; err != nil {
			tx.Rollback()
			return fmt.Errorf("fail to insert role, err:", err)
		}
		luckDraw := &entity.LuckyDraw{
			Role_id: role.ID,
		}
		if err := tx.Create(luckDraw).Error; err != nil {
			tx.Rollback()
			return fmt.Errorf("fail to insert lucky draw, err:", err)
		}
		if err := tx.Commit().Error; err != nil {
			return fmt.Errorf("fail to commit transaction, err: %v", err)
		}
	}

	return nil
}

func initPool(address string, maxIdle, maxActive int, IdleTimeout time.Duration) {
	RedisPool = &redis.Pool{
		MaxIdle:     maxIdle,
		MaxActive:   maxActive,
		IdleTimeout: IdleTimeout,
		Dial: func() (redis.Conn, error) {
			return redis.Dial("tcp", address, redis.DialDatabase(0))
		},
	}
}

func InitRedis() {
	initPool(viper.GetString("redis.dns"), viper.GetInt("redis.maxIdle"),
		viper.GetInt("redis.maxActive"), time.Duration(viper.GetInt("redis.idleTimeout")))

	// 测试需要~
	conn := RedisPool.Get()
	_, err := conn.Do("FLUSHALL")
	if err != nil {
		fmt.Printf("清空 Redis 数据失败: %v\n", err)
	} else {
		fmt.Println("Redis 数据已清空")
	}
}

// redis过期键监听
func ListenRedisExpiredEvents() {
	conn := RedisPool.Get()
	psc := redis.PubSubConn{Conn: conn}

	err := psc.PSubscribe("__keyevent@0__:expired")
	if err != nil {
		log.Fatal("PSubscribe error:", err)
	}

	go func() {
		for {
			switch v := psc.Receive().(type) {
			case redis.PMessage: // 一定要是PMessage啊！！！！！！
				key := string(v.Data)
				fmt.Println("🔔 Redis 到期键:", key)
				if key == "floor_rewards_cache" {
					NotifyingDelete(27106)
				} else if key == "cumulative_rewards_cache" {
					NotifyingDelete(27107)
				}
			case redis.Subscription:
				fmt.Printf("Subscribed to %s: %s (%d)\n", v.Kind, v.Channel, v.Count)
			case error:
				log.Println("Redis pubsub error:", v)
				return
			default:
				fmt.Println("Redis PubSub Receive:", v)
			}
		}
	}()
}

//// v1.0版本 直接轮询发布缓存失效
//func NotifyingDelete(code int64) {
//	// 轮询正在连接的客户端，发布本地缓存失效的消息
//	for conn, role := range OnlineUserMgr.OnlineUsers {
//		resMes := &entity.Msg{
//			Code: code,
//		}
//		data, err := json.Marshal(resMes)
//		if err != nil {
//			fmt.Printf("客户端 RoleId=%d 通知失败，原因：%s\n", role.ID, err)
//		}
//		tf := &util.Transfer{
//			Conn: conn,
//		}
//		err = tf.WritePkg(data)
//		if err != nil {
//			fmt.Printf("客户端 RoleId=%d 通知失败，原因：%s\n", role.ID, err)
//		}
//	}
//}

func NotifyingDelete(code int64) {
	ch, err := MQ.Channel()
	if err != nil {
		fmt.Println("MQ Channel error:", err)
		return
	}
	defer ch.Close()

	for _, role := range OnlineUserMgr.OnlineUsers {
		msg := map[string]interface{}{
			"role_id":   role.ID,
			"code":      code,
			"try_count": 0,
		}
		body, _ := json.Marshal(msg)
		err = ch.Publish(
			"",                 // exchange
			"notifying_delete", // routing key
			false,              // mandatory
			false,              // immediate
			amqp.Publishing{
				ContentType: "application/json",
				Body:        body,
			},
		)
		if err != nil {
			fmt.Printf("RoleId = %d 通知消息入队失败:%v\n", role.ID, err)
		}
	}

}

// 初始化 RabbitMQ 连接
func InitRabbitMQ() {
	var err error
	url := viper.GetString("rabbitmq.dns")
	MQ, err = amqp.Dial(url)
	if err != nil {
		log.Fatalf("无法连接到 RabbitMQ: %v", err)
	}
	log.Println("成功连接到 RabbitMQ")

	// retry → 回流 user_notify_queue
	argsRetry := amqp.Table{
		"x-dead-letter-exchange":    "",
		"x-dead-letter-routing-key": "user_notify_queue",
		"x-message-ttl":             int32(5000),
	}
	ch, err := MQ.Channel()
	if err != nil {
		log.Fatalf("MQ Channel error: %v", err)
	}
	_, err = ch.QueueDeclare("user_notify_retry", true, false, false, false, argsRetry)
	if err != nil {
		return
	}

}

// 更新抽奖排行榜
func UpdateLuckyDrawRanking(roleID uint, consumedTokens int64, drawNums int64) error {
	conn := RedisPool.Get()
	defer conn.Close()

	key := fmt.Sprintf("lucky_draw:rank_tokens")
	_, err := conn.Do("ZADD", key, consumedTokens, roleID)

	if err != nil {
		return err
	}

	key2 := fmt.Sprintf("lucky_draw:rank_drawNums")
	_, err = conn.Do("ZADD", key2, drawNums, roleID)

	return err
}

func ConsumeRankingUpdates() {
	ch, _ := MQ.Channel()
	q, _ := ch.QueueDeclare("ranking_update", true, false, false, false, nil)
	msgs, _ := ch.Consume(q.Name, "", true, false, false, false, nil)

	for msg := range msgs {
		var update entity.LuckyDrawRank
		if err := json.Unmarshal(msg.Body, &update); err != nil {
			log.Println("解析消息失败:", err)
			continue
		}

		// 执行 Redis 排行榜更新
		err := UpdateLuckyDrawRanking(update.Role_id, update.Draw_num, update.Consumed_tokens)
		if err != nil {
			fmt.Println("update mq error:", err)
		}
	}
}

func ConsumeNotifyingDelete() {
	ch, _ := MQ.Channel()
	q, _ := ch.QueueDeclare("notifying_delete", true, false, false, false, nil)
	msgs, _ := ch.Consume(
		q.Name, "", false,
		false, false, false, nil,
	)

	go func() {
		for d := range msgs {
			var m struct {
				RoleID   uint  `json:"role_id"`
				Code     int64 `json:"code"`
				TryCount int   `json:"try_count"`
			}
			err := json.Unmarshal(d.Body, &m)
			if err != nil {
				d.Ack(false)
				continue
			}

			conn := OnlineUserMgr.OnlineIdConn[m.RoleID]
			if conn == nil {
				fmt.Printf("RoleId=%d 不在线，跳过\n", m.RoleID)
				d.Ack(false)
				continue
			}

			resMes := &entity.Msg{Code: m.Code}
			data, _ := json.Marshal(resMes)
			tf := &transfer.Transfer{Conn: conn}
			err = tf.WritePkg(data)
			if err != nil {
				m.TryCount++
				if m.TryCount <= 3 {
					// 重试入队
					chRetry, _ := MQ.Channel()
					body, _ := json.Marshal(m)
					_ = chRetry.Publish(
						"", "user_notify_retry", false, false,
						amqp.Publishing{
							ContentType: "application/json",
							Body:        body,
						},
					)
					fmt.Printf("RoleId=%d 通知失败，重试第 %d 次\n", m.RoleID, m.TryCount)
				} else {
					fmt.Printf("RoleId=%d 通知失败超过限制，丢弃\n", m.RoleID)
				}
			}

			d.Ack(false)
		}
	}()
}

// 奖励增量更新 异步推送
func ConsumeRewards() {
	ch, _ := MQ.Channel()
	q, _ := ch.QueueDeclare("rewards_update", true, false, false, false, nil)
	msgs, _ := ch.Consume(q.Name, "", true, false, false, false, nil)

	for msg := range msgs {
		var update entity.ItemForQueue
		if err := json.Unmarshal(msg.Body, &update); err != nil {
			log.Println("解析消息失败:", err)
			continue
		}
		RoleId := update.Role_id
		conn := OnlineUserMgr.OnlineIdConn[RoleId]
		// 增量推送
		resMes := entity.Msg{
			Code:          27201,
			Status:        true,
			Repsonse_data: string(msg.Body),
		}
		data, err := json.Marshal(resMes)
		if err != nil {
			fmt.Println("msg fail:", msg)
			continue
		}
		tf := &transfer.Transfer{
			Conn: conn,
		}
		err = tf.WritePkg(data)
		if err != nil {
			fmt.Println("msg fail:", msg)
			continue
		}
	}
}
